package com.xiaofu.live.user.provider.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MQProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.*;

@Configuration
@Slf4j
public class RocketMQProducerConfig {

    @Autowired
    private RocketMQProducerProperties rocketMQProducerProperties;

    @Value("${spring.application.name}")
    private String applicationName;

    @Bean
    public MQProducer rocketMQProducer(RocketMQConsumerProperties rocketMQConsumerProperties) {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(100, 150, 3, TimeUnit.MINUTES,
                new ArrayBlockingQueue<>(1000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName(applicationName + ":rmp-producer:"+ ThreadLocalRandom.current().nextInt(1000));
                return thread;
            }
        });
        DefaultMQProducer producer = new DefaultMQProducer();
        try {
            producer.setNamesrvAddr(rocketMQProducerProperties.getNameSrv());
            producer.setProducerGroup(rocketMQProducerProperties.getGroupName());
            producer.setRetryTimesWhenSendFailed(rocketMQProducerProperties.getRetryTimes());
            producer.setRetryTimesWhenSendAsyncFailed(rocketMQProducerProperties.getRetryTimes());
            producer.setRetryAnotherBrokerWhenNotStoreOK(true);
            producer.setAsyncSenderExecutor(threadPoolExecutor);
            producer.start();
            log.info("mq生产者启动成功, nameSrv: {}",rocketMQConsumerProperties.getNameSrv());
        } catch (MQClientException e) {
            throw new RuntimeException(e);
        }

        return producer;
    }
}
